/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package aiaudio.database;

import aiaudio.database.tables.TableStructure;
import aiaudio.lastfm.hbase.CannotCreateTableException;
import java.io.IOException;
import java.util.Calendar;
import java.util.HashMap;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;

/**
 *
 * @author nastya
 */
public class DatabaseStorage {

    private Configuration hadoop;
    private HBaseAdmin database;
    private String databasePrefix;

    private HashMap<String, TableDescriptor> existingTables = new HashMap<String, TableDescriptor>();
   
    public DatabaseStorage(Configuration hadoop, HBaseAdmin database, String databasePrefix) {
        this.hadoop = hadoop;
        this.database = database;
        this.databasePrefix = databasePrefix;
    }

    public void removeAllTemporaryTables() throws IOException{
        removeAllTablesWithPrefix("temporary.*");
    }
    
    public void removeAllDatabaseTablesWithKey(String key) throws IOException{
        removeAllTablesWithPrefix(createDatabaseName(key) + ".*");
    }

    public TableDescriptor getByName(String alias){
        return existingTables.get(alias);
    }
    
    private String createDatabaseName(String key) {
        return databasePrefix + "_" + key;
    }
    
    private void removeAllTablesWithPrefix(String prefix) throws IOException{
        database.disableTables(prefix);
        database.deleteTables(prefix);
    }
    public String createTemporaryMatrixTable() throws IOException{
        String tableDatabaseName = "temporaryMatrix" + Calendar.getInstance().getTimeInMillis(); 
        HTable table = createTableInDb(tableDatabaseName, true, TableStructure.TemporaryMatrix);
        addToStorage(table, tableDatabaseName, TableStructure.TemporaryMatrix, existingTables, tableDatabaseName);
        return tableDatabaseName;
    }
    
    public String createTemporaryTable() throws IOException{
        String tableDatabaseName = "temporary" + Calendar.getInstance().getTimeInMillis(); 
        HTable table = createTableInDb(tableDatabaseName, true, TableStructure.Temporary);
        addToStorage(table, tableDatabaseName, TableStructure.Temporary, existingTables, tableDatabaseName);
        return tableDatabaseName;
    }
    
    public void createTable(String tableAlias, TableStructure structure, boolean dropIfExists) throws CannotCreateTableException, IOException{
        String databaseTableName = createDatabaseName(tableAlias);
        HTable table = createTableInDb(databaseTableName, dropIfExists, structure);
        addToStorage(table, databaseTableName, structure, existingTables, tableAlias);
    }
    

    protected void addToStorage(HTable table, String databaseTableName, TableStructure structure, HashMap<String, TableDescriptor> tableList, String tableAlias) throws IOException {
        TableDescriptor descriptor = new TableDescriptor(databaseTableName, table, structure);
        tableList.put(tableAlias, descriptor);
    }

    protected HTable createTableInDb(String databaseTableName, boolean dropIfExists, TableStructure structure) throws IOException {
        if (database.tableExists(databaseTableName)) {
            if (dropIfExists) {
                database.disableTable(databaseTableName);
                database.deleteTable(databaseTableName);
            }
            createTableWithFamilies(databaseTableName, structure.getAllFamilies());
        }
        return new HTable(database.getConfiguration(), databaseTableName);
    }
    
    private void createTableWithFamilies(String tableName, Family... families) throws IOException {
        HTableDescriptor tableDescriptor = new HTableDescriptor(tableName);

        for (Family f : families) {
            HColumnDescriptor descriptor = new HColumnDescriptor(f.getName());
            tableDescriptor.addFamily(descriptor);
        }

        database.createTable(tableDescriptor);
    }
    
}
